package cn.iocoder.yudao.module.gpt.service.chat;

import cn.hutool.cache.Cache;
import cn.hutool.cache.CacheUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.util.CharUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.framework.security.core.LoginUser;
import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
import cn.iocoder.yudao.module.gpt.api.dto.GPTUserMessageDTO;
import cn.iocoder.yudao.module.gpt.api.dto.MQResponseMessageDTO;
import cn.iocoder.yudao.module.gpt.config.ChatGPTProperties;
import cn.iocoder.yudao.module.gpt.dal.redis.chatsession.ChatSessionRedisDO;
import cn.iocoder.yudao.module.gpt.dal.redis.chatsession.ChatSessionRedisDao;
import cn.iocoder.yudao.module.gpt.listener.GPTStreamResponseMessageDTO;
import com.unfbx.chatgpt.OpenAiStreamClient;
import com.unfbx.chatgpt.entity.chat.ChatCompletion;
import com.unfbx.chatgpt.entity.chat.Message;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import org.jetbrains.annotations.Nullable;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

@Slf4j
@EnableConfigurationProperties(ChatGPTProperties.class)
@Service
public class GPTChatServiceImpl implements GPTChatService {

    @Resource
    private ChatGPTProperties chatGPTProperties;

    @Resource
    private StreamBridge streamBridge;
    @Resource
    private OpenAiStreamClient client;
    @Resource
    private ChatSessionRedisDao chatSessionRedisDao;

    private static final String GPTResponseBinding = "gptResponse-flux-out";

    private final Cache<String,String> fifoCache = CacheUtil.newFIFOCache(4096, DateUnit.MINUTE.getMillis());

    @Override
    public void getGPTAnswerWithFlux(@NotNull GPTUserMessageDTO userMessageDTO) {
        List<ChatSessionRedisDO> chatSessionRedisHistory = chatSessionRedisDao.searchHistory(userMessageDTO.getSessionId());
        List<Message> messages = new ArrayList<>();
        for(ChatSessionRedisDO chatSessionRedisDO : chatSessionRedisHistory){
            Message message = Message.builder().role(chatSessionRedisDO.getUserRole()).content(chatSessionRedisDO.getContent()).build();
            messages.add(message);
        }
        Message userMessage = Message.builder().role(Message.Role.USER).content(userMessageDTO.getContent()).build();
        messages.add(userMessage);
        //聊天模型：gpt-3.5
        ChatCompletion chatCompletion = ChatCompletion.builder().user(userMessageDTO.getName()).model(StrUtil.isEmpty(chatGPTProperties.getModel()) ? ChatCompletion.Model.GPT_3_5_TURBO.getName() : chatGPTProperties.getModel()).n(chatGPTProperties.getAnswerCount()).temperature(chatGPTProperties.getTemperature()).messages(messages).build();
        client.streamChatCompletion(chatCompletion, new EventSourceListener() {
            @Override
            public void onClosed(@org.jetbrains.annotations.NotNull EventSource eventSource) {
                super.onClosed(eventSource);
                log.info("OpenAI关闭sse连接...");
            }

            @Override
            public void onEvent(@org.jetbrains.annotations.NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @org.jetbrains.annotations.NotNull String data) {
                super.onEvent(eventSource, id, type, data);
                MQResponseMessageDTO mqResponseMessageDTO = null;
                if (data.equals("[DONE]")) {
                    log.info("OpenAI返回数据结束了");
                    finishChatContent(userMessageDTO);
                    mqResponseMessageDTO = generateMQResponseMessage(data,userMessageDTO.getSessionId());
                }else{
                    GPTStreamResponseMessageDTO responseMessage = JSONUtil.toBean(data, GPTStreamResponseMessageDTO.class);
                    if(CollUtil.isNotEmpty(responseMessage.getChoices())){
                        GPTStreamResponseMessageDTO.Choice choice = responseMessage.getChoices().get(0);
                        if(ObjectUtil.isNotNull(choice) && ObjectUtil.isNotNull(choice.getDelta())){
                            GPTStreamResponseMessageDTO.Choice.Delta delta = choice.getDelta();
                            if(ObjectUtil.isEmpty(delta)){
                                delta = new GPTStreamResponseMessageDTO.Choice.Delta();
                                delta.setContent("[DONE]");
                                finishChatContent(userMessageDTO);
                            }else{
                                boolean empty = fifoCache.isEmpty();
                                if(empty){
                                    fifoCache.put(userMessageDTO.getSessionId(),delta.getContent());
                                }
                                String s = fifoCache.get(userMessageDTO.getSessionId());
                                fifoCache.put(userMessageDTO.getSessionId(),StrUtil.appendIfMissing(s,delta.getContent()));
                            }
                            mqResponseMessageDTO = generateMQResponseMessage(delta.getContent(),userMessageDTO.getSessionId());
                        }
                    }
                }
                if(ObjectUtil.isNotNull(mqResponseMessageDTO))
                    streamBridge.send(GPTResponseBinding, mqResponseMessageDTO);
            }

            @Override
            public void onFailure(@org.jetbrains.annotations.NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
                super.onFailure(eventSource, t, response);
                if(Objects.isNull(response)){
                    log.error("OpenAI  sse连接异常:{}", ObjectUtil.isNotNull(t) ? t.getMessage() : "");
                    return;
                }
                ResponseBody body = response.body();
                if (Objects.nonNull(body)) {
                    try {
                        log.error("OpenAI  sse连接异常data：{}，异常：{}", body.string(), ObjectUtil.isNotNull(t) ? t.getMessage() : "");
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                } else {
                    log.error("OpenAI  sse连接异常data：{}，异常：{}", response, ObjectUtil.isNotNull(t) ? t.getMessage() : "");
                }
                eventSource.cancel();
            }

            @Override
            public void onOpen(@org.jetbrains.annotations.NotNull EventSource eventSource, @org.jetbrains.annotations.NotNull Response response) {
                super.onOpen(eventSource, response);
                log.info("OpenAI建立sse连接...");
            }
        });
    }
    private MQResponseMessageDTO generateMQResponseMessage(String content,  String sessionId){
        MQResponseMessageDTO mqResponseMessageDTO = new MQResponseMessageDTO();
        mqResponseMessageDTO.setContent(content);
        mqResponseMessageDTO.setSessionId(sessionId);
        return mqResponseMessageDTO;
    }

    private void finishChatContent(@NotNull GPTUserMessageDTO userMessageDTO){
        String finishedContent = fifoCache.get(userMessageDTO.getSessionId());
        ChatSessionRedisDO chatSessionRedisDO = new ChatSessionRedisDO();
        chatSessionRedisDO.setUserRole(Message.Role.ASSISTANT);
        chatSessionRedisDO.setContent(finishedContent);
        chatSessionRedisDao.insertHistory(chatSessionRedisDO,userMessageDTO.getSessionId());
    }
    @Override
    public String validateSessionId(String sessionId) {
        List<ChatSessionRedisDO> chatSessionRedisDOS = chatSessionRedisDao.searchHistory(sessionId);
        if(CollUtil.isNotEmpty(chatSessionRedisDOS)){
            return "";
        }
        ChatSessionRedisDO chatSessionRedisDO = new ChatSessionRedisDO();
        chatSessionRedisDO.setUserRole(Message.Role.USER);
        chatSessionRedisDO.setContent("[START]");
        chatSessionRedisDao.insertHistory(chatSessionRedisDO,sessionId);
        return sessionId;
    }

}
